#include <string.h>
#include <pthread.h>
#include <set>

#include "threadpool.h"

#define THREAD_ADJUST_INTERVAL 2 // 调整线程数量步长

struct threadpool_t
{
    bool _shutdown;

    threadpool_task_t *_task_queue; // 任务队列
    size_t _queue_size;             // 队列总容量[定值]
    size_t _task_num;               // 任务数量
    int _task_front;                // 队头下标
    int _task_rear;                 // 队尾下标

    size_t _min_thread_num;       // 最小线程数量
    size_t _max_thread_num;       // 最大线程数量
    size_t _live_thread_num;      // 存活线程数量
    size_t _busy_thread_num;      // 忙线程数量
    size_t _exit_thread_num;      // 需要销毁的线程数
    std::set<pthread_t> _threads; // 线程号集合(工作线程),用set容器方便管理线程号
    pthread_t _tid_ajust;

    pthread_mutex_t _mutex_pool;     // 池结构体的锁
    pthread_mutex_t _mutex_busy_num; // 忙碌线程的锁

    pthread_cond_t _cond_produce; // 生产产品
    pthread_cond_t _cond_consum;  // 消费产品
};

threadpool_t *threadpool_create(size_t qsize, size_t minthnum, size_t maxthnum)
{
    // 创建结构体并初始化
    threadpool_t *pool = NULL;

    do
    {
        if (!(pool = new threadpool_t))
            break;

        pool->_shutdown = false;
        pool->_queue_size = qsize;
        pool->_task_num = 0;
        pool->_task_front = 0;
        pool->_task_rear = 0;
        pool->_min_thread_num = minthnum;
        pool->_max_thread_num = maxthnum;
        pool->_live_thread_num = 0;
        pool->_busy_thread_num = 0;
        pool->_exit_thread_num = 0;

        if (pthread_mutex_init(&pool->_mutex_pool, NULL) ||
            pthread_cond_init(&pool->_cond_produce, NULL) ||
            pthread_cond_init(&pool->_cond_consum, NULL))
            break;

        // 创建任务队列
        if (!(pool->_task_queue = new threadpool_task_t[pool->_queue_size]))
            break;

        // 按照最小线程数创建工作线程
        pthread_t tid_new;
        for (int i = 0; i < pool->_min_thread_num; i++)
        {
            if (!pthread_create(&tid_new, NULL, worker_thread, pool))
            {
                pool->_threads.insert(tid_new);
                pool->_live_thread_num++;
            }
            else
            {
                perror("create thread error");
                exit(1);
            }
        }

        // 创建管理者线程
        if (pthread_create(&pool->_tid_ajust, NULL, adjust_thread, pool))
        {
            perror("create thread error");
            exit(1);
        }

        return pool;

    } while (0);

    if (pool)
    {
        if (pool->_task_queue)
            delete[] pool->_task_queue;

        delete pool;
    }
    return nullptr;
}

void threadpool_destroy(threadpool_t *pool)
{
    if (!pool)
        return;

    // 诱杀
    pool->_shutdown = true;
    pthread_cond_broadcast(&pool->_cond_produce);

    // 等待线程推出
    pthread_join(pool->_tid_ajust, NULL);
    for (auto thr : pool->_threads)
    {
        pthread_join(thr, NULL);
    }

    // 销毁互斥量之前要先lock
    pthread_mutex_lock(&pool->_mutex_pool);
    pthread_mutex_destroy(&pool->_mutex_pool);
    pthread_cond_destroy(&pool->_cond_produce);
    pthread_cond_destroy(&pool->_cond_consum);

    if (pool->_task_queue)
        delete[] pool->_task_queue;

    delete pool;

    std::cout << "threadpool is destroied..." << std::endl;
}

void *worker_thread(void *arg)
{
    threadpool_t *pool = (threadpool_t *)arg;

    while (1)
    {
        pthread_mutex_lock(&pool->_mutex_pool);
        while (pool->_task_num == 0 && !pool->_shutdown)
        {
            pthread_cond_wait(&pool->_cond_produce, &pool->_mutex_pool);

            // 如果当前有要销毁的线程就退出
            if (pool->_exit_thread_num > 0)
            {
                pool->_exit_thread_num--;
                if (pool->_live_thread_num > pool->_min_thread_num)
                {
                    pool->_threads.erase(pthread_self());
                    pool->_live_thread_num--;
                    pthread_mutex_unlock(&pool->_mutex_pool);

                    pthread_detach(pthread_self());
                    pthread_exit(NULL);
                }
            }
        }

        if (pool->_shutdown)
        {
            pthread_mutex_unlock(&pool->_mutex_pool);
            pthread_exit(NULL);
        }

        // 从任务队列取一个任务，头指针后移一位,任务数减1
        if (pool->_task_num == 0)
        {
            std::cerr << "task queue error!" << std::endl;
            exit(1);
        }
        threadpool_task_t task = pool->_task_queue[pool->_task_front];
        pool->_task_front = (pool->_task_front + 1) % pool->_queue_size;
        pool->_task_num--;
        pthread_mutex_unlock(&pool->_mutex_pool);

        pthread_cond_signal(&pool->_cond_consum);

        // 忙线程数加1
        pthread_mutex_lock(&pool->_mutex_busy_num);
        pool->_busy_thread_num++;
        pthread_mutex_unlock(&pool->_mutex_busy_num);

        // 执行任务
        task.callback(task.arg1, task.arg2);

        // 忙线程数减1
        pthread_mutex_lock(&pool->_mutex_busy_num);
        pool->_busy_thread_num--;
        pthread_mutex_unlock(&pool->_mutex_busy_num);
    }

    return nullptr;
}

void *adjust_thread(void *arg)
{
    threadpool_t *pool = (threadpool_t *)arg;
    if (!pool)
        return nullptr;

    while (!pool->_shutdown)
    {
        sleep(2);

        static double last_busy_num = 0;

        // 获取任务队列的任务数量
        pthread_mutex_lock(&pool->_mutex_pool);
        int task_num = pool->_task_num;
        int live_thr_num = pool->_live_thread_num; // 调整前的线程数

        // 获取忙碌线程的数量
        pthread_mutex_lock(&pool->_mutex_busy_num);
        int busy_thr_num = pool->_busy_thread_num;
        pthread_mutex_unlock(&pool->_mutex_busy_num);
        pthread_mutex_unlock(&pool->_mutex_pool);

        // 由于busy_thr_num不稳定，以busy_thr_num加权平均值来评估线程池
        double average_busy_num = (last_busy_num + busy_thr_num) / 2;
        last_busy_num = average_busy_num;

        // std::cout << "task_num: " << task_num << std::endl;
        // std::cout << "busy_thr_num: " << busy_thr_num << std::endl;
        // std::cout << "average_busy_thr_num: " << average_busy_num << std::endl;
        // std::cout << "live_thread_num: " << live_thr_num << std::endl;

        // 当加权忙碌线程占总线程大于80%时，增加线程
        if (average_busy_num * 1.25 > live_thr_num &&
            live_thr_num < pool->_max_thread_num)
        {
            pthread_t tid_new;

            pthread_mutex_lock(&pool->_mutex_pool);
            for (int i = 0;
                 i < THREAD_ADJUST_INTERVAL && // 将线程增加THREAD_ADJUST_INTERVAL个
                 pool->_live_thread_num < pool->_max_thread_num;
                 i++)
            {
                pthread_create(&tid_new, NULL, worker_thread, pool);
                pool->_threads.insert(tid_new);
                pool->_live_thread_num++;
            }
            pthread_mutex_unlock(&pool->_mutex_pool);
        }

        // 当忙碌的存活线程占比低于20%时，减少线程
        if (average_busy_num * 5 < live_thr_num &&
            live_thr_num > pool->_min_thread_num)
        {
            pthread_mutex_lock(&pool->_mutex_pool);
            // 减少三分之一线程
            int exit_num = pool->_exit_thread_num = pool->_live_thread_num / 3; 
            pthread_mutex_unlock(&pool->_mutex_pool);

            for (int i = 0; i < exit_num; i++)
                pthread_cond_signal(&pool->_cond_produce);
        }
    }

    return nullptr;
}

void threadpool_addtask(void (*callback)(int, void *), int arg1, void *arg2, threadpool_t *pool)
{
    if (!pool)
        return;

    pthread_mutex_lock(&pool->_mutex_pool);

    while (pool->_task_num >= pool->_queue_size && !pool->_shutdown)
        pthread_cond_wait(&pool->_cond_consum, &pool->_mutex_pool);

    if (pool->_shutdown)
    {
        pthread_mutex_unlock(&pool->_mutex_pool);
        return;
    }

    if (pool->_task_num >= pool->_queue_size)
    {
        std::cerr << "task queue error!" << std::endl;
        exit(1);
    }
    pool->_task_queue[pool->_task_rear].callback = callback;
    pool->_task_queue[pool->_task_rear].arg1 = arg1;
    pool->_task_queue[pool->_task_rear].arg2 = arg2;
    pool->_task_rear = (pool->_task_rear + 1) % pool->_queue_size;
    pool->_task_num++;

    pthread_mutex_unlock(&pool->_mutex_pool);

    pthread_cond_signal(&pool->_cond_produce);
}